use chrono::{Datelike, NaiveDate};
use clickhouse_rs::types::SqlType;

use crate::base::{get_ch_pool,Table};
pub struct MonthTrlsProc<'a> {
    tconf: &'a Table,
    columns: &'a Vec<(String, SqlType)>,
    tmp_table: &'a str,
}
impl<'a> MonthTrlsProc<'a> {
    pub fn new(tconf: &'a Table, columns: &'a Vec<(String, SqlType)>, tmp_table: &'a str) -> Self {
        Self {
            tconf,
            columns,
            tmp_table,
        }
    }
    ///折叠指定日期的数据，整体清0
    async fn collapsing_data(
        &self,
        date: &NaiveDate,
        date_col: &str,
        cols_str1: &str,
        cols_str2: &str,
    ) -> anyhow::Result<()> {
        let date = date.format("%Y-%m-%d").to_string();
        let key_cols_str = self.tconf.unique_keys.join(",");
        let sql =
            format!(r#"insert into {}({cols_str1},DIVI.DPID,DIVI.EMID,DIVI.DSVL,SIGN)
            select {cols_str2},anyLast(DIVI.DPID),anyLast(DIVI.EMID),anyLast(DIVI.DSVL),-1 from {}
            where {date_col}=toDate('{date}')
            group by {key_cols_str} having sum(SIGN)>0"#,self.tconf.table_name,self.tconf.table_name);
        let mut client = get_ch_pool().get_handle().await?;
        client.execute(sql).await?;
        Ok(())
    }
    ///运行数据处理
    pub async fn run(&self, date: &NaiveDate) -> anyhow::Result<()> {
        let (date_col, cols_str1, cols_str2) = self.month_trls_to_tmp(date).await?;
        self.collapsing_data(date, date_col, &cols_str1, &cols_str2)
            .await?;
        //临时表数据汇总
        let key_cols_str = self.tconf.unique_keys.join(",");
        let sql =format!("CREATE TABLE IF NOT EXISTS {}_2 AS {}",self.tmp_table,self.tmp_table);
        let mut client = get_ch_pool().get_handle().await?;
        client.execute(sql).await?;
        let sql =format!("TRUNCATE TABLE {}_2",self.tmp_table);
        client.execute(sql).await?;
        let sql =
        format!("insert into {}_2({cols_str1},SIGN) select {cols_str2},1 from {} group by {key_cols_str}",self.tmp_table,self.tmp_table);
        client.execute(sql).await?;
        let sql =format!("DROP TABLE {}",self.tmp_table);
        client.execute(sql).await?;
        let sql =format!("RENAME TABLE {}_2 TO {}",self.tmp_table,self.tmp_table);
        client.execute(sql).await?;
        Ok(())
    }
    ///之前的月交易流水插入临时表
    async fn month_trls_to_tmp(&self, date: &NaiveDate) -> anyhow::Result<(&str, String, String)> {
        let mut date_col: Option<&str> = None;
        let mut cols_str1 = String::new();
        let mut cols_str2 = String::new();
        let date_str = date.format("%Y-%m-%d").to_string();
        for (cname, st) in self.columns.iter() {
            if cname == "SIGN" || cname.starts_with("DIVI.") {
                continue;
            }
            cols_str1.push_str(cname);
            if !self.tconf.unique_keys.contains(cname) {
                match st {
                    SqlType::UInt32
                    | SqlType::UInt64
                    | SqlType::UInt128
                    | SqlType::Int32
                    | SqlType::Int64
                    | SqlType::Int128
                    | SqlType::Float32
                    | SqlType::Float64
                    | SqlType::Decimal(_, _) => {
                        if self.tconf.columns.contains_key(cname){
                            let cexp = format!("sum(SIGN*{cname})");
                            cols_str2.push_str(&cexp);
                        }else{
                            let cexp=format!("max({})",cname);
                            cols_str2.push_str(&cexp)
                        }
                    }
                    SqlType::Date => {
                        date_col = {
                            let cexp = format!("toDate('{date_str}')");
                            cols_str2.push_str(&cexp);
                            Some(cname.as_str())
                        }
                    }
                    _ => {
                        let exp=format!("max({})",cname);
                        cols_str2.push_str(&exp)
                    },
                }
            } else {
                cols_str2.push_str(cname);
            }
            cols_str1.push(',');
            cols_str2.push(',');
        }
        if date_col.is_none() {
            return Err(anyhow::anyhow!(
                "导入文件{}类型为MonthTrls,字段中必须有日期列",
                self.tconf.file_name
            ));
        }
        cols_str1.pop();
        cols_str2.pop();
        let date_col = date_col.unwrap();
        if date.day() == 1 {
            return Ok((date_col, cols_str1, cols_str2));
        }
        let sdate = NaiveDate::from_ymd_opt(date.year(), date.month(), 1).unwrap();
        let sdate = sdate.format("%Y-%m-%d").to_string();
        let key_cols_str = self.tconf.unique_keys.join(",");
        let sql =
            format!(r#"insert into {}({cols_str1},SIGN) select {cols_str2},-1 from {}
            where {date_col}>=toDate('{sdate}') and {date_col}<toDate('{date_str}')
            group by {key_cols_str} having sum(SIGN)>0"#,self.tmp_table,self.tconf.table_name);
        let mut client = get_ch_pool().get_handle().await?;
        client.execute(sql).await?;
        return Ok((date_col, cols_str1, cols_str2));
    }
}
